package cn.com.zetatech.maple.core.task;

import cn.com.zetatech.maple.annotation.WriterHandler;
import cn.com.zetatech.maple.core.utils.BeanUtils;
import cn.com.zetatech.maple.core.utils.ConfigurationManager;
import cn.com.zetatech.maple.entity.*;
import cn.com.zetatech.maple.parser.parser.impl.DeviceLogFileParser;
import cn.com.zetatech.maple.parser.utils.ScriptEnginePool;
import cn.com.zetatech.maple.rule.RuleParser;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import javax.script.Bindings;
import javax.script.ScriptEngine;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author cheng.cao@zetatech.com.cn
 */
@NoArgsConstructor
@AllArgsConstructor
@Slf4j
@SuppressWarnings("all")
public class ParseTask implements Callable<List<Map<String, String>>> {
    private Pattern pattern;
    private FtpDFile ftpDFile;

    @Override
    public List<Map<String, String>> call() throws Exception {
        try {
            String remote = ftpDFile.getRemote();
            if (!ftpDFile.isSuccess()) {
                log.error("该日志文件为下载失败的日志文件: {}，跳过", remote);
                return Lists.newArrayList();
            }
            Matcher m = pattern.matcher(remote);
            List<String> keys = new ArrayList<>();
            while (m.find() && keys.size() < 2) {
                keys.add(m.group().replaceAll("\\d+", ""));
            }
            String key = Joiner.on("_").join(keys);
            Rule rule = RuleParser.getFields(key);
            if (rule == null) {
                log.error("文件{}未找到对应的解析规则，忽略!", remote);
                return Lists.newArrayList();
            }
            List<Field> fields = rule.getFields();
            List<ColumnDef> columnDefs = fields.stream()
                    .filter(f -> !f.isIgnore())
                    .map(f -> new ColumnDef(f.getColumn(), f.getType()))
                    .collect(Collectors.toList());
            DeviceLogFileParser parser = new DeviceLogFileParser(BeanUtils.getBean(rule.getClazz()));
            // 本地处理
            if (rule.getLocalHandler() != null) {
                rule.getLocalHandler().handle(ftpDFile);
            }
            List<Map<String, String>> datas = parser.parse(new File(ftpDFile.getLocal()), rule);
            log.info(remote + " ==> " + datas.size());

            boolean writerEnable = ConfigurationManager.me().getBoolean("system.writer.enable", true);
            if (writerEnable) {
                Writer writer = rule.getWriter();
                Class clazz = writer.getClass();
                WriterHandler annoWH = (WriterHandler) clazz.getAnnotation(WriterHandler.class);
                boolean needPartition = annoWH.needPartition();

                // 调用脚本设置分区
                if (needPartition) {
                    java.lang.reflect.Field field = clazz.getDeclaredField("partitions");
                    field.setAccessible(true);
                    List<HiveWriter.Partition> partitions = (List<HiveWriter.Partition>) field.get(writer);
                    if (CollectionUtils.isNotEmpty(partitions)) {
                        for (HiveWriter.Partition p : partitions) {
                            if (StringUtils.isNotBlank(p.getValue())) {
                                continue;
                            }
                            if (StringUtils.isBlank(p.getScript())) {
                                continue;
                            }
                            ScriptEngine engine = ScriptEnginePool.getEngine(p.getEngineName());
                            Bindings bindings = engine.createBindings();
                            bindings.put("file", ftpDFile.getLocal());
                            bindings.put("datas", datas);
                            try {
                                String processResult = (String) engine.eval(p.getScript(), bindings);
                                p.setValue(processResult);
                            } catch (Exception e) {
                                log.error("设置分区{}异常", p.getName(), e);
                            }
                        }
                    }
                }
                cn.com.zetatech.maple.writer.WriterHandler handler = BeanUtils.getBean(annoWH.value());
                handler.write(rule.getWriter(), columnDefs, datas);
            } else {
                for (Map<String, String> data : datas) {
                    log.info("{}", data);
                }
            }
            try {
                log.info("删除本地文件:{}", ftpDFile.getLocal());
                new File(ftpDFile.getLocal()).delete();
            } catch (Exception e) {
                log.error("删除本地文件:{}失败!", ftpDFile.getLocal(), e);
            }
            return datas;
        } catch (Exception e) {
            log.error("处理文件:{}失败,本地文件将保留!", ftpDFile.getLocal(), e);
        }

        return Lists.newArrayList();
    }

}




